package com.xr.redis.pubsub.service.impl;

import com.xr.redis.pubsub.model.Message;
import com.xr.redis.pubsub.service.IPubSubServiceAdapter;
import com.xr.redis.pubsub.service.listener.AbstractPubSubListener;
import com.xr.redis.util.JSONUtil;
import com.xr.redis.util.ThreadPoolUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by Administrator on 2018/8/9.
 */
public class ListImplementAdapter implements IPubSubServiceAdapter {

  protected Logger logger = LoggerFactory.getLogger(ListImplementAdapter.class);

  private ConcurrentHashMap<String, List<AbstractPubSubListener>> channelMapToListener;

  private Jedis jedisClient;

  private int bufferSize = 100000; // default value
  private ArrayBlockingQueue<Message> localCacheBuffer = null;

  public ListImplementAdapter(Jedis jedisClient, ConcurrentHashMap<String, List<AbstractPubSubListener>> channelMapToListener) {
    this.jedisClient = jedisClient;
    this.channelMapToListener = channelMapToListener;
  }

  @Override
  public void init() throws Exception {
    localCacheBuffer = new ArrayBlockingQueue<Message>(bufferSize);

    // 拉取消息
    fetchMessage();

    // 分发消息进行处理
    dispatcheMessage();
  }

  @Override
  public void publish(String channel, Message message) throws Exception {
    if (channel != null && message != null) {
      message.setChannel(channel);
      this.jedisClient.lpush(getKey(), JSONUtil.toJSONString(message));
    }
  }

  @Override
  public void subscribe(String channel) throws Exception {

  }

  /**
   * 两个线程，同时拉去前后两个list的日志
   * <p>
   * 每次上传日志都会修改一次list的过期时间
   * list 最长有效期15min（10min的间隔 + 5min有效期）
   * <p>
   * 前面一个list，getLastKey()，在后面一个list创建5min后还没有处理完的日志将被丢弃
   * 后面一个list，getKey()
   */
  public void fetchMessage() {
    new Thread(new Runnable() {
      @Override
      public void run() {
        while (true) {
          try {
            Message message = JSONUtil.parseObject(jedisClient.rpop(getLastKey()), Message.class);
            offerLogItem(message);
          } catch (Exception e) {
            logger.error("拉取前一个list日志信息异常：", e);
          }
        }
      }
    }).start();

    new Thread(new Runnable() {
      @Override
      public void run() {
        while (true) {
          try {
            Message message = JSONUtil.parseObject(jedisClient.rpop(getKey()), Message.class);
            offerLogItem(message);
          } catch (Exception e) {
            logger.error("拉取当前list日志信息异常：", e);
          }
        }
      }
    }).start();

    logger.info("开始拉取日志信息...");
  }

  private void offerLogItem(Message message) {
    if (message != null) {
      // 队列未满时，返回true；队列满时返回false。
      if (!localCacheBuffer.offer(message)) {
        logger.warn("队列满，丢弃当前日志，缓冲区大小：" + localCacheBuffer.size());
      }
      logger.info("接收日志 id:" + message.getMsgId());
    }
  }

  private void dispatcheMessage() {
    new Thread(new Runnable() {
      @Override
      public void run() {
        try {
          // 循环读取本地日志缓冲区
          logger.info("开始处理日志信息...");
          while (true) {
            final Message message = localCacheBuffer.take();
            if (message != null) {
              ThreadPoolUtil.submit(new Runnable() {
                @Override
                public void run() {
                  try {
                    final List<AbstractPubSubListener> listeners = channelMapToListener.get(message.getChannel());
                    if (listeners != null && listeners.size() > 0) {
                      logger.info("处理日志 id：" + message.getMsgId());
                      for (AbstractPubSubListener apsl : listeners) {
                        apsl.onMessage(message);
                      }
                    }
                  } catch (Exception e) {
                  }
                }
              });
            }
          }
        } catch (InterruptedException e) {
          logger.error("分发日志信息异常：", e);
        }
      }
    }).start();
  }

  /**
   * 每10min为一个key
   * 将当前时间格式化为：yyyyMMddHHmm，然后最后一位替换为0
   *
   * @return
   */
  private String getKey() {
    String key = parseDateToString(curentTimeaddMinute(0), "yyyyMMddHHmm");
    return key.substring(0, 11) + "0";
  }

  /**
   * 上一个十分钟的key（当前key - 10min）
   *
   * @return
   */
  private String getLastKey() {
    String key = parseDateToString(curentTimeaddMinute(-10), "yyyyMMddHHmm");
    return key.substring(0, 11) + "0";
  }

  public static String parseDateToString(Date date, String pattern){
    return new SimpleDateFormat(pattern).format(date);
  }

  public Date curentTimeaddMinute(int term){
    Calendar date = Calendar.getInstance();
    date.add(Calendar.MINUTE, term);
    return date.getTime();
  }
}
